package com.corn.flink.lesson2;

import cn.hutool.core.io.FileUtil;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.Arrays;

/**
 * @author : Jim Wu
 * @version 1.0
 * @function :
 * @since : 2023/2/23 17:37
 */

public class FlinkTypeHintsDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        String filePath = FileUtil.getAbsolutePath("classpath:input/wordcount.txt");
        DataStreamSource<String> ds = env.readTextFile(filePath);

        ds.flatMap((String data, Collector<Tuple2<String, Integer>> collector) -> {
            String[] word = data.split(" ");
            Arrays.stream(word).forEach(w -> {
                collector.collect(Tuple2.of(w, 1));
            });
        })
                // TypeHint 或者 Types
                .returns(new TypeHint<Tuple2<String, Integer>>() {
                })
                .keyBy(data -> data.f0)
                .sum(1)
                .print();

        env.execute();
    }
}
